package com.open.source;

import com.open.source.kafka.ApplicationContextHolder;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import lombok.SneakyThrows;
import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.DescribeConsumerGroupsResult;
import org.apache.kafka.clients.admin.MemberDescription;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author ZonLen since on 2022/9/6 下午3:37
 */
@RestController
@RequestMapping("admin")
public class KafkaAdminController {

  @Autowired
  private KafkaProperties kafkaProperties;

  @SneakyThrows
  @GetMapping("consumer")
  public Boolean consumer() {
    final String groupId = kafkaProperties.buildConsumerProperties().get(
        ConsumerConfig.GROUP_ID_CONFIG).toString();
    final DescribeConsumerGroupsResult consumerGroups = Admin
        .create(kafkaProperties.buildAdminProperties())
        .describeConsumerGroups(Collections.singleton(groupId));
    final Collection<MemberDescription> members = consumerGroups.describedGroups().get(groupId)
        .get().members();
    final String topicName = ApplicationContextHolder.getEnvironment()
        .resolvePlaceholders(MqConstant.TAX_TOPIC_DELAY_DEMO);
    return members.stream().anyMatch(
        memberDescription -> memberDescription.assignment().topicPartitions().stream().anyMatch(
            topicPartition -> topicPartition.topic()
                .equalsIgnoreCase(topicName)));

  }
}
